SpringBoot整合RabbitMQ进阶（手动确认消息）

2023-07-15 16:43| 来源: 网络整理| 查看: 265

文章目录 1、回调函数2、消息确认3、代码实现一下手动确认

在上一篇中我们已经讲解了SpringBoot整合RabbitMQ的基础知识，这里附上链接，SpringBoot整合RabbitMQ基础部分，如果对SpringBoot整合RabbitMQ不熟悉的同学，可以先看完这一篇文章在看进阶部分的文章。下面我们在来讲解一下SpringBoot整合RabbitMQ进阶部分。

这里写的东西都是基于上篇文章来写的，上篇文章已经写的代码，这里就不在重复。

1、回调函数在rabbitmq-provider项目的application.yml文件上，加上消息确认的配置项 server: port: 8021 spring: #配置项目名称 application: name: rabbitmq-provider #配置rabbitmq 服务器 rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest # 确认消息发送到交换机上 publisher-confirm-type: correlated # 确认消息已发送到队列 publisher-returns: true 再配置相关的确认消息回调函数，这个类是个配置类 @Configuration public class RabbitConfig { /** * 自定义RabbitTemplate * @return */ @Bean public RabbitTemplate createRabbitTempalte(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); //设置开启Manatory，才能触发回调函数，无论消息推送结果怎么样都会强制调用回调函数 rabbitTemplate.setMandatory(true); // 设置确认发送到交换机的回调函数 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean b, String s) { System.out.println("ConfirmCallback: "+"相关数据："+correlationData); System.out.println("ConfirmCallback: "+"确认情况："+b); System.out.println("ConfirmCallback: "+"原因："+s); } }); //设置确认消息已发送到队列的回调 rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() { @Override public void returnedMessage(ReturnedMessage returnedMessage) { System.out.println("消息："+returnedMessage.getMessage()); System.out.println("回应码："+returnedMessage.getReplyCode()); System.out.println("回应信息："+returnedMessage.getReplyText()); System.out.println("交换机："+returnedMessage.getExchange()); System.out.println("路由键："+returnedMessage.getRoutingKey()); } }); return rabbitTemplate; } }

可以看到上面写了两个回调函数，一个叫 setConfirmCallback ，一个叫 setRetrunCallback；那么以上这两种回调函数都是在什么情况会触发呢？

先从总体的情况分析，推送消息存在四种情况：

①消息推送到server，但是在server里找不到交换机 ②消息推送到server，找到交换机了，但是没找到队列 ③消息推送到sever，交换机和队列啥都没找到 ④消息推送成功

下面我们来分别写接口来测试上面的四种情况 ①消息推送到server，但是在server里找不到交换机写个测试接口，把消息推送到名为‘non-existent-exchange’的交换机上（这个交换机是没有创建没有配置的）： @GetMapping("/TestMessageAck") public String TestMessageAck() { String messageId = String.valueOf(UUID.randomUUID()); String messageData = "message: non-existent-exchange test message "; String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); Map map = new HashMap(); map.put("messageId", messageId); map.put("messageData", messageData); map.put("createTime", createTime); rabbitTemplate.convertAndSend("non-existent-exchange", "testDirectRouting", map); return "ok"; }

请求接口查看控制台打印情况

ConfirmCallback: 相关数据：null ConfirmCallback: 确认情况：false ConfirmCallback: 原因：channel error; protocol method: #method(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost '/', class-id=60, method-id=40)

结论： ①这种情况触发的是 ConfirmCallback 回调函数。

②消息推送到server，找到交换机了，但是没找到队列这种情况就是需要新增一个交换机，但是不给这个交换机绑定队列，我来简单地在DirectRabitConfig里面新增一个直连交换机，名叫‘lonelyDirectExchange’，但没给它做任何绑定配置操作：

@Bean public DirectExchange lonelyDirectExchange(){ return new DirectExchange("lonelyDirectExchange"); } @GetMapping("/TestMessageAck2") public String TestMessageAck2() { String messageId = String.valueOf(UUID.randomUUID()); String messageData = "message: lonelyDirectExchange test message "; String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); Map map = new HashMap(); map.put("messageId", messageId); map.put("messageData", messageData); map.put("createTime", createTime); rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map); return "ok"; }

调用接口查看控制台打印情况

ConfirmCallback: 相关数据：null ConfirmCallback: 确认情况：true ConfirmCallback: 原因：null 消息：(Body:'{createTime=2020-12-30 22:41:23, messageId=cdd69332-7446-4cc2-945f-3a7972523fe7, messageData=message: lonelyDirectExchange test message }' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0]) 回应码：312 回应信息：NO_ROUTE 交换机：lonelyDirectExchange 路由键：TestDirectRouting

可以看到这种情况，两个函数都被调用了；这种情况下，消息是推送成功到服务器了的，所以ConfirmCallback对消息确认情况是true；而在RetrunCallback回调函数的打印参数里面可以看到，消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE 。结论：②这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数

③消息推送到sever，交换机和队列啥都没找到这种情况其实一看就觉得跟①很像，没错，③和①情况回调是一致的，所以不做结果说明了。结论： ③这种情况触发的是 ConfirmCallback 回调函数。

④消息推送成功那么测试下，按照正常调用之前消息推送的接口就行，就调用下 /sendFanoutMessage接口，可以看到控制台输出：

ConfirmCallback: 相关数据：null ConfirmCallback: 确认情况：true ConfirmCallback: 原因：null

结论： ④这种情况触发的是 ConfirmCallback 回调函数。

2、消息确认

和生产者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。所以，消息接收的确认机制主要存在三种模式：

①自动确认，这也是默认的消息确认情况。 AcknowledgeMode.NONE RabbitMQ成功将消息发出（即将消息成功写入TCP Socket）中立即认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。所以这种情况如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。一般这种情况我们都是使用try catch捕捉异常后，打印日志用于追踪数据，这样找出对应数据再做后续处理。

② 根据情况确认，这个不做介绍 ③ 手动确认，这个比较关键，也是我们配置接收消息确认机制时，多数选择的模式。消费者收到消息后，手动调用basic.ack/basic.nack/basic.reject后，RabbitMQ收到这些消息后，才认为本次投递成功。 basic.ack用于肯定确认 basic.nack用于否定确认（注意：这是AMQP 0-9-1的RabbitMQ扩展） basic.reject用于否定确认，但与basic.nack相比有一个限制:一次只能拒绝单条消息

消费者端以上的3个方法都表示消息已经被正确投递，但是basic.ack表示消息已经被正确处理。而basic.nack,basic.reject表示没有被正确处理：着重讲下reject，因为有时候一些场景是需要重新入列的。 channel.basicReject(deliveryTag, true); 拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行。下次不想再消费这条消息了。使用拒绝后重新入列这个确认模式要谨慎，因为一般都是出现异常的时候，catch异常再拒绝入列，选择是否重入列。但是如果使用不当会导致一些每次都被你重入列的消息一直消费-入列-消费-入列这样循环，会导致消息积压。顺便也简单讲讲 nack，这个也是相当于设置不消费某条消息。

channel.basicNack(deliveryTag, false, true);

第一个参数依然是当前消息到的数据的唯一id; 第二个参数是指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。第三个参数是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。

同样使用不确认后重新入列这个确认模式要谨慎，因为这里也可能因为考虑不周出现消息一直被重新丢回去的情况，导致积压。

3、代码实现一下手动确认

在消费者rabbitmq-customer中加入 MessageListenerConfig 配置类

@Configuration public class MessageListenerConfig { @Resource private CachingConnectionFactory cachingConnectionFactory; @Resource private MyAckReceiver myAckReceiver; /** * @return * @Date 2020/12/30 16:21 * @Author jiangheng * @Description //TODO 自定义监听容器，不使用@RabbitListen 注解来消费消息，而是使用自定义的SimpleMessageListenerContainer来消费消息 **/ @Bean public SimpleMessageListenerContainer simpleMessageListenerContainer() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(cachingConnectionFactory); //设置最小并发的消费者数量 container.setConcurrentConsumers(1); //设置最大并发的消费者数量 container.setMaxConcurrentConsumers(20); //限流，单位时间内消费多少条记录 container.setPrefetchCount(10); //设置rabbit 确认消息的模式，默认是自动确认 container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置一个队列 //container.setQueueNames("testDirectQueue"); //监听队列名称 container.setQueueNames("testDirectQueue"); //设置消息监听类 container.setMessageListener(myAckReceiver); return container; } }

对应的手动确认消息监听类，MyAckReceiver.java（手动确认模式需要实现 ChannelAwareMessageListener）：之前的相关监听器可以先注释掉，以免造成多个同类型监听器都监听同一个队列。这里的获取消息转换，只作参考，如果报数组越界可以自己根据格式去调整。

@Component public class MyAckReceiver implements ChannelAwareMessageListener { /** * 处理单个队列 * @param message * @param * @throws IOException */ @Override public void onMessage(Message message, Channel channel) throws IOException { long deliveryTag = message.getMessageProperties().getDeliveryTag(); //因为传递消息的时候使用map 传递的，所以将Map从Message内取出做处理 String msg = message.toString(); //从源码中可以发现他是用 , 隔开的，所以我们用 , 进行拆分成数组 String msgArray [] = msg.split(","); Map msgMap = mapStringToMap(msgArray[1].trim(),3); String messageId=msgMap.get("messageId"); String messageData=msgMap.get("messageData"); String createTime=msgMap.get("createTime"); System.out.println(" MyAckReceiver messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); System.out.println("消费的主题消息来自："+message.getMessageProperties().getConsumerQueue()); //第二个参数手动确认可以被批量处理，当参数为true时，则可以一次性确认delivery_tag小于等于传入值的所有消息 try { channel.basicAck(deliveryTag, true); //第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝 //channel.basicReject(deliveryTag, true); } catch (IOException e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } } @Override public void onMessage(Message message) { } @Override public void onMessageBatch(List messages, Channel channel) { } private Map mapStringToMap(String str,int entryNum) { str = str.substring(1, str.length() - 1); String[] strs = str.split(",",entryNum); Map map = new HashMap(); for (String string : strs) { String key = string.split("=")[0].trim(); String value = string.split("=")[1]; map.put(key, value); } return map; } }

这时，先调用接口/sendDirectMessage，给直连交换机TestDirectExchange 的队列TestDirectQueue 推送一条消息，可以看到监听器正常消费了下来：

消费的消息来自的队列名为：testDirectQueue 消息成功消费到 messageId:4d88d3fd-8f68-489a-8c9d-87069257ddde messageData:test message, hello! createTime:2020-12-30 22:57:06 执行TestDirectQueue中的消息的业务处理流程...... simpleMessageListenerContainer-1

手动处理多个队列的消息这里只需要基于上面的代码，两步就可以实现。第一步，在MessageListenerConfig #simpleMessageListenerContainer方法中添加你要监听的队列名就可以了，只需要改下面一行代码就可以了。

//还可以设置多个队列，用逗号分开你要监听的队列名 container.setQueueNames("testDirectQueue", "fanout.A");

第二部，修改 MyAckReceiver ，在方法上加上判断就可以了。

/** * 处理多个队列 * @param message * @param channel * @throws Exception */ @Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理 String msg = message.toString(); //可以点进Message里面看源码,单引号直接的数据就是我们的map消息数据 String[] msgArray = msg.split("'"); Map msgMap = mapStringToMap(msgArray[1].trim(),3); String messageId=msgMap.get("messageId"); String messageData=msgMap.get("messageData"); String createTime=msgMap.get("createTime"); if ("testDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){ System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue()); System.out.println("消息成功消费到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); System.out.println("执行TestDirectQueue中的消息的业务处理流程......"); System.out.println(Thread.currentThread().getName()); } if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){ System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue()); System.out.println("消息成功消费到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); System.out.println("执行fanout.A中的消息的业务处理流程......"); } channel.basicAck(deliveryTag, true); // channel.basicReject(deliveryTag, true);//为true会重新放回队列 } catch (Exception e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } }

以上内容来之这位博主的分享，非常感谢

【本文地址】

公司简介

联系我们